package cn.hy.dataSync.task.runner;


import cn.hy.dataSync.config.*;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.utils.RdbUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;


/**
 * 写 sql 线程,
 */
@Slf4j
public class WriterRunner extends AbstractRunner implements Runnable {

    // 接受者 通道
    private RecordReceiver recordReceiver;

    public WriterRunner(TaskConfig taskConfig) {
        super(taskConfig);
    }

    public void setRecordReceiver(RecordReceiver receiver) {
        this.recordReceiver = receiver;
    }

    @Override
    public void run() {
        Validate.isTrue(this.recordReceiver != null);

        try {
            log.debug("task writer starts to do init ...");
            init(); // 空实现

            log.debug("task writer starts to do prepare ...");
            prepare(); // 空实现

            log.debug("task writer starts to write ...");
            startWrite(recordReceiver);

            log.debug("task writer starts to do post ...");
            post(); // 空实现

            super.markSuccess();
        } catch (Throwable e) {
            log.error("Writer Runner Received Exceptions:", e);
            super.markFail(e);
        } finally {
            log.debug("task writer starts to do destroy ...");
            super.destroy();
        }
    }

    @Override
    public void shutdown() {
        recordReceiver.shutdown();
    }

    private void init() {

    }

    private void prepare() {

    }

    private void startWrite(RecordReceiver recordReceiver) {
        WriterConfig writerConfig = this.getTaskConfig().getJobConfig().getWriterConfig();
        Connection connection = RdbUtils.getConnection(DataBaseType.MySql,
                writerConfig.getJdbcUrl(), writerConfig.getUsername(), writerConfig.getPassword());

        int batchSize = writerConfig.getBatchSize();
        List<Record> writeBuffer = new ArrayList<>(batchSize);
        try {
            Record record;
            while ((record = recordReceiver.getFromReader()) != null) {
                writeBuffer.add(record);
                if (writeBuffer.size() >= batchSize) {
                    doBatchInsert(connection, writeBuffer);
                    writeBuffer.clear();
                }
            }
            if (!writeBuffer.isEmpty()) {
                doBatchInsert(connection, writeBuffer);
                writeBuffer.clear();
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            writeBuffer.clear();
            RdbUtils.close(connection);
        }
    }

    private void post() {

    }

    private void doBatchInsert(Connection connection, List<Record> buffer)
            throws SQLException {
        PreparedStatement preparedStatement = null;
        try {
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement(this.getTaskConfig().getWriterSql());

            for (Record record : buffer) {
                fillPreparedStatement(preparedStatement, record);
                preparedStatement.addBatch(); // 添加到批量提交
            }
            preparedStatement.executeBatch();
            connection.commit();
        } catch (SQLException e) {
            // 出现异常，一条一条写入
            log.warn("回滚此次写入, 采用每次写入一行方式提交. 因为:" + e.getMessage());
            connection.rollback();
            doOneInsert(connection, buffer);
        } catch (Exception e) {
            // 可以获取此次一条写入失败的原因
            log.error("单次写入失败：",e);
            throw new RuntimeException(e);
        } finally {
            RdbUtils.close(preparedStatement);
        }
    }

    private void doOneInsert(Connection connection, List<Record> buffer) {
        PreparedStatement preparedStatement = null;
        try {
            connection.setAutoCommit(true);
            preparedStatement = connection.prepareStatement(this.getTaskConfig().getWriterSql());

            for (Record record : buffer) {
                try {
                    fillPreparedStatement(preparedStatement, record);
                    preparedStatement.execute();
                } catch (SQLException e) {
                    log.debug(e.toString());
                    Optional.ofNullable(getTaskDirtyCollector()).ifPresent(
                            collector -> collector.collectDirtyRecord(record, e));
                } finally {
                    // 最后不要忘了关闭 preparedStatement
                    preparedStatement.clearParameters();
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            RdbUtils.close(preparedStatement);
        }
    }

    private void fillPreparedStatement(PreparedStatement preparedStatement, Record record)
            throws SQLException {
        List<Transformer> transformers = this.getTaskConfig()
                .getJobConfig().getWriterConfig().getTransformers();
        for (int i = 1; i <= transformers.size(); i++) {
            Transformer transformer = transformers.get(i - 1);
            preparedStatement.setObject(i, record.get(transformer.getColumn()));
        }
    }
}
